#pragma once

#include "../mq_comm/mq_helper.hpp"
#include "../mq_comm/mq_log.hpp"
#include "../mq_comm/mq_msg.pb.h"

#include <iostream>
#include <unordered_map>
#include <mutex>
#include <vector>
#include <string>
#include <memory>
#include <list>
namespace mq
{
    using namespace ns_helper;
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"

    using MessagePtr = std::shared_ptr<mq::MQMessage>;

    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname) : _qname(qname)
        {
            if (basedir.back() != '/')
                basedir.push_back('/');
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
            if (FileHelper(basedir).IsExist() == false)
            {
                assert(FileHelper::createDirectory(basedir));
            }
            createMsgFile();
        }
        bool createMsgFile()
        {
            if (FileHelper(_datafile).IsExist())
            {
                return true;
            }
            bool ret = FileHelper::createFile(_datafile);
            if (!ret)
            {
                LOG(ERROR) << "创建队列文件失败: " << _datafile << std::endl;
                return false;
            }
            return true;
        }
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }
        bool Insert(MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }
        bool remove(const MessagePtr &msg)
        {
            // 将有效位 置为 "0" ;  修改payload中的字段或者反序列化时使用mutable_payload
            msg->mutable_payload()->set_valid("0");
            // 对数据进行序列化
            std::string msg_str = msg->payload().SerializeAsString();
            // 将数据写入到指定位置
            if (msg_str.size() != msg->len())
            {
                LOG(WARNING) << "未能将数据写入, 因为新数据与与原数据长度不一致!\n";
                return false;
            }
            FileHelper fhelper(_datafile);
            bool ret = fhelper.Write(msg_str.c_str(), msg->offset(), msg_str.size());
            if (ret == false)
            {
                LOG(ERROR) << "队列消息数据写入失败!\n";
                return false;
            }
            return true;
        }
        std::list<MessagePtr> gc() // 垃圾回收
        {
            bool ret;
            std::list<MessagePtr> result;
            ret = load(result);
            if (ret == false)
            {
                LOG(ERROR) << "加载有效数据失败！\n";
                return result;
            }
            LOG(DEBUG) << "垃圾回收后, 有效消息的数量: " << result.size() << std::endl;
            // 2. 将有效数据，进行序列化存储到临时文件中
            FileHelper::createFile(_tmpfile);
            for (auto &msg : result)
            {
                LOG(DEBUG) << "向临时文件写入数据: " << msg->payload().body() << std::endl;
                ret = insert(_tmpfile, msg);
                if (ret == false)
                {
                    LOG(ERROR) << "向临时文件写入消息数据失败！！\n";
                    return result;
                }
            }
            LOG(DEBUG) << "垃圾回收后, 向临时文件写入数据完毕, 临时文件大小: " << FileHelper(_tmpfile).Size() << std::endl;
            // // 3. 删除源文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                LOG(ERROR) << "删除源文件失败！\n";
                return result;
            }
            // 4. 修改临时文件名，为源文件名称
            ret = FileHelper(_tmpfile).Rename(_datafile);
            if (ret == false)
            {
                LOG(ERROR) << "修改临时文件名称失败！\n";
                return result;
            }
            // 5. 返回新的有效数据
            return result;
        }

    private:
        // 加载读取文件数据中的有效数据
        bool load(std::list<MessagePtr> &result)
        {
            FileHelper data_file(_datafile);
            size_t msg_size = 0, offset = 0;
            size_t fsize = data_file.Size();
            // std::cout << "mq_message->load: offset: " << offset << " fsize: " << fsize << std::endl;
            bool ret;
            // 一条一条的读取数据, 并判断数据是否有效
            while(offset < fsize)
            {
                // 先读取4字节 -->读取消息长度
                ret = data_file.Read((char*)&msg_size, offset, sizeof(size_t));
                if (ret == false)
                {
                    LOG(ERROR) << "队列消息长度读取失败! \n";
                    return false;
                }
                offset += sizeof(size_t);
                
                // std::cout << "message load msg_size: " << msg_size << std::endl;
                std::string msg(msg_size, '\0');

                data_file.Read(&msg[0], offset, msg_size);

                if (!ret)
                {
                    LOG(ERROR) << "队列消息读取失败! \n";
                    return false;
                }
                offset += msg_size;

                // 进行反序列化, 判断消息是否有效
                MessagePtr msgp = std::make_shared<MQMessage>();
                msgp->mutable_payload()->ParseFromString(msg);
                if (msgp->payload().valid() == "0")
                {
                    // 消息无效
                    LOG(DEBUG) << "加载的消息无效: " << msgp->payload().body() << std::endl;
                    continue;
                }
                // 保存有效数据
                result.push_back(msgp);
            }
            return true;
        }
        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 新增数据都是添加在文件末尾的
            // 1. 进行消息的序列化，获取到格式化后的消息
            std::string body = msg->payload().SerializeAsString();
            // 2. 获取文件长度
            FileHelper helper(filename);
            size_t fsize = helper.Size();
            size_t msg_size = body.size();
            // 写入逻辑：1. 先写入4字节数据长度， 2， 再写入指定长度数据
            bool ret = helper.Write((char *)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                LOG(ERROR) << "向队列数据文件写入数据长度失败！\n";
                return false;
            }
            // 3. 将数据写入文件的指定位置
            ret = helper.Write(body.c_str(), fsize + sizeof(size_t), body.size());
            if (ret == false)
            {
                LOG(ERROR) << "向队列数据文件写入数据失败！";
                return false;
            }
            // 4. 更新msg中的实际存储信息
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_len(body.size());
            return true;
        }

    private:
        std::string _datafile;
        std::string _tmpfile;
        std::string _qname;
    };
    // 队列中消息的管理
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        // 创建/打开队列数据文件, 恢复队列历史数据
        QueueMessage(std::string &basedir, const std::string &qname)
            : _mapper(basedir, qname),
              _qname(qname), _valid_count(0), _total_count(0)
        {
        }
        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _msgs = _mapper.gc();   // 垃圾回收(文件中), 返回有效数据
            for (auto &msg : _msgs) // 更新内存中, 持久化消息属性
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _valid_count = _total_count = _msgs.size();
            return true;
        }
        // BasicProperties消息属性 body消息内容
        // 分为两种: 1.不写入到文件; 2.写入到文件
        bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 构建消息对象
            MessagePtr msg = std::make_shared<MQMessage>();
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            std::unique_lock<std::mutex> lock(_mtx);
            // 判断是否需要持久化
            if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                msg->mutable_payload()->set_valid("1"); // 将有效位置为1
                bool ret = _mapper.Insert(msg);         // 进行持久化操作
                if (!ret)
                {
                    LOG(ERROR) << "持久化存储消息失败: " << body << std::endl;
                    return false;
                }
                _valid_count += 1;
                _total_count += 1;
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _msgs.push_back(msg);
            return true;
        }
        // 根据消息id进行删除, 每次删除后需要判断是否进行垃圾回收
        // 取出消息后,从待确认的hash中移除
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            // 在待确认的hash中寻找消息
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                LOG(DEBUG) << "没有找到要删除的消息!\n";
                return true;
            }
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                _mapper.remove(it->second); // 这里并没有真正的删除, 而是把有效位置为1;
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;
                // 判断并进行垃圾回收
                gc();
            }
            _waitack_msgs.erase(msg_id);

            return true;
        }
        // 获取队首消息 (从_msgs中取)
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            if (_msgs.size() == 0)
            {
                return MessagePtr();
            }
            // 从链表中取出消息
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 把取出的消息加入到待确认hash中
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }
        // 删除队列所有消息
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }
        // 获取待推送(可获取的)消息数量
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _msgs.size();
        }
        // 获取待确认消息数量
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _waitack_msgs.size();
        }
        // 获取有效持久化消息数量
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _valid_count;
        }
        // 获取所有持久化消息数量
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _total_count;
        }

    private:
        bool GCCheck()
        {
            // 总消息数量大于2000, 有效数据占比不足 50% 进行垃圾回收
            if (_total_count > 2000 && (_valid_count * 10) / _total_count < 5)
            {
                return true;
            }
            return false;
        }
        void gc()
        {
            // 判断是否需要垃圾回收
            if (GCCheck() == false)
                return;
            std::list<MessagePtr> msgs = _mapper.gc(); // 进行垃圾回收操作, 返回有效数据的链表
            // 更新有效消息在内存中的属性
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    LOG(WARNING) << "垃圾回收后, 有一条消息在内存中没有进行管理!\n";
                    // 重新将消息加入到(待推送的)list中;
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    continue;
                }
                it->second->set_offset(msg->offset());
                it->second->set_len(msg->len());
            }
            //3. 更新当前的有效消息数量 & 总的持久化消息数量
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::mutex _mtx;
        std::string _qname;                                        // 队列名称
        MessageMapper _mapper;                                     // 消息持久化管理
        std::list<MessagePtr> _msgs;                               // 待推送消息的链表
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息
        // 为了方便查找持久化的消息，_durable_msgs和_msgs中的MessagePtr指向同一个message对象，进行垃圾回收后需要更新内存中消息的属性；
        // 为什么要更新内存中待确认消息的属性？
        // 消息推送到消费客户端，消费客户端进行消息确认，确认后需要移除消息，但是这里的移除消息并不是将消息移除，而是将消息有效位置为1；
        // 那么就需要找到该消息在文件中的偏移量（使用内存中消息的属性），所以这里必须要更新；
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息，推送一条消息后，就把该消消息对象的messagePtr加入到待确认的消息集合中；
        //移除消息时需要从待确认的hash映射中取messagePtr（指向内存中的消息对象）；

        size_t _valid_count; // 有效数据数量
        size_t _total_count; // 总体数据数量
    };
    // 对外总体消息的管理
    // 注意在编写时锁重复保护的问题
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string &basedir) : _basedir(basedir) {}
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &qmsg : _queue_msgs)
            {
                qmsg.second->clear();
            }
        }
        // 初始化一个队列
        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                {
                    return;
                }
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            //如果在QueueMessage对象恢复数据, 就会导致构建QueueMessage对象时速度较慢, 并且也会导致锁重复保护
            // 将恢复历史数据的接口单独封装, 这样可以加快对象的构建
            qmp->recovery();
        }
        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    return;
                }
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            qmp->clear();
        }
        bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    LOG(ERROR) << qname << "队列新增消息失败：没有找到消息管理句柄!";
                    return false;
                }
                qmp = it->second;
            }

            return qmp->insert(bp, body, queue_is_durable);
        }
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    LOG(ERROR) << qname << "获取队列队首消息失败：没有找到消息管理句柄!\n";
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->front();
        }
        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    LOG(ERROR) << qname << "确认队列消息: " << msg_id << " 失败：没有找到消息管理句柄!";
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
            return;
        }
        size_t getable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    LOG(ERROR) << qname << "获取可确认队列消息失败, 没有找到消息管理句柄!";
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }
        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    LOG(ERROR) << qname << "获取全部持久化消息数量失败, 没有找到消息管理句柄!";
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    LOG(ERROR) << qname << "获取有效持久化消息数量失败, 没有找到消息管理句柄!";
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    LOG(ERROR) << qname << "获取待确认消息数量失败, 没有找到消息管理句柄!";
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }

    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; // 队列名称和队列消息的映射
    };
}
